﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using PServiceBus.Core.Interface;
using PServiceBus.Core.Manager;
using PServiceBus.Core.Runtime;
using ThreadTask = System.Threading.Tasks.Task;
using PServiceBus.Core.Provider;
using PServiceBus.Core.Logger;
using PServiceBus.Core.Runtime.Messages;
using PServiceBus.Core.Runtime.Configuration;
using System.Diagnostics;
using PServiceBus.Core.Runtime.Transports;
using PServiceBus.Core.Runtime.Topics;
using PServiceBus.Core.Runtime.Extensions;

namespace PServiceBus.Core.Components {
    public class MessageRedispatcher : ComponentBase {

        private static IObjectManager<FailedMessage> _failedMessages;

        private static IObjectManager<FailedMessage> FailedMessages {
            get {
                if (_failedMessages == null)
                    _failedMessages = ObjectManager<FailedMessage>.Create(ConfigurationProvider.ComponentContainerConfig.Address);
                return _failedMessages;
            }
        }

        private static IObjectManager<DeadMessage> _deadMessages;

        private static IObjectManager<DeadMessage> DeadMessages {
            get {
                if (_deadMessages == null)
                    _deadMessages = ObjectManager<DeadMessage>.Create(ConfigurationProvider.ComponentContainerConfig.Address);
                return _deadMessages;
            }
        }

        private static object _lockObject = new object();

        #region IComponent Members

        public override void Execute() {
            var retryMessageInterval =
                (ConfigurationProvider.ServiceBusConfig.RetryMessageInterval == TimeSpan.Zero ?
                TimeSpan.FromSeconds(5) : ConfigurationProvider.ServiceBusConfig.RetryMessageInterval).TotalMilliseconds;
            var failedMessages = FailedMessages;
            var subscribersDict = SubscriptionCollection.Instance.SubscriptionDict.SelectMany(x => x.Value);
            var subscribers = new Dictionary<Guid, ISubscriber>();
            foreach (var kv in subscribersDict)
                subscribers[kv.Key] = kv.Value;
            var retryList = new List<FailedMessage>();
            var deadList = new List<DeadMessage>();
            var task = new List<ThreadTask>();
            var publishedMessages = new List<PublishedMessage>();

            foreach (var message in failedMessages) {
                if (!message.IsValid) continue;
                var retryMessage = new FailedMessage(message.Message, message.ID, message.TopicID) { RetryCount = message.RetryCount - 1 };
                if (DateTime.Now.Subtract(message.CreateDate).TotalMilliseconds < retryMessageInterval) {
                    retryMessage.RetryCount = message.RetryCount;
                    retryMessage.Responses = message.Responses;
                    retryMessage.CreateDate = message.CreateDate;
                    retryList.Add(retryMessage);
                    continue;
                }
                var publishedMessage = new PublishedMessage(message.Message, retryMessage.TopicID);
                foreach (var response in message.Responses) {
                    var subscriber = subscribers.Select(response.SubscriberID);
                    if (subscriber == null) continue;
                    task.Add(ThreadTask.Factory.StartNew(o =>
                    {
                        var sub = o as ISubscriber;
                        var emptyGuid = Guid.Empty;
                        var transports = sub.Transports.Where(t => t.TopicID == message.TopicID || t.TopicID == emptyGuid);
                        var subscription = subscriber.Subscriptions.FirstOrDefault(s => s.ID == message.TopicID);
                        foreach (var transportInfo in transports) {
                            var transport = transportInfo.Transport;
                            if (transport.GetType().Name != response.TransportResponse.Type) continue;
                            var tranMessage = TransportProvider.FilterMessage(sub, message.TopicID, message.Message);
                            if (tranMessage == null) continue;
                            if (subscription.ID != Guid.Empty && subscription.NeedMessageHeaders)
                                tranMessage.IncludeHeaderInMessages();
                            if (tranMessage.Parameters.Count > 0) {
                                var tranResponse = transport.Send(TransportProvider.TransformMessage(transport.Format, tranMessage),
                                    null);
                                var messageResponse = new MessageResponse() { SubscriberName = sub.Name, SubscriberID = sub.ID, TransportResponse = tranResponse };
                                lock (_lockObject) {
                                    if (!tranResponse.Success) retryMessage.Responses.Add(messageResponse);
                                    else _totalRetriedMessages++;
                                    publishedMessage.Responses.Add(messageResponse);
                                }
                            }
                        }
                    }, subscriber));
                }
                MethodHelper.TryLog(() => ThreadTask.WaitAll(task.ToArray()));
                if (retryMessage.IsValid) {
                    retryList.Add(retryMessage);
                    _totalRetriedMessages++;
                }
                if (retryMessage.IsDead) {
                    deadList.Add(new DeadMessage(retryMessage));
                    _totalDeadMessages++;
                }
                publishedMessage.Type = retryMessage.IsDead ? MessageType.Dead : MessageType.Retry;
                publishedMessages.Add(publishedMessage);
            }
            failedMessages.AttachMany(retryList);
            DeadMessages.AttachMany(deadList);
            foreach (var message in publishedMessages) message.Save();
            _perfCounter.SetValue(TOTAL_DEAD_MESSAGES, _totalDeadMessages);
            _perfCounter.SetValue(TOTAL_RETRIED_MESSAGES, _totalRetriedMessages);
            retryList = null;
            deadList = null;
            publishedMessages = null;
        }

        public override void Close() {
        }

        public override void Init() {
            _perfCounter.Load(ConfigurationProvider.ComponentConfig.Name, new Dictionary<string, System.Diagnostics.PerformanceCounterType>() { 
               {TOTAL_DEAD_MESSAGES, PerformanceCounterType.NumberOfItems64},
               {TOTAL_RETRIED_MESSAGES, PerformanceCounterType.NumberOfItems64}
            });
        }

        const string TOTAL_DEAD_MESSAGES = "TotalDeadMessages";
        const string TOTAL_RETRIED_MESSAGES = "TotalRetriedMessages";

        private long _totalDeadMessages = 0;
        private long _totalRetriedMessages = 0;
        private PerfCounter _perfCounter = new PerfCounter();
        #endregion
    }
}
